package io.sundial.balance.impl;

import com.google.common.base.Splitter;
import io.sundial.balance.Balancer;
import io.sundial.balance.exception.BalancingException;
import io.sundial.executor.Executor;
import io.sundial.job.JobSharding;
import io.sundial.task.Task;
import io.sundial.util.StrKit;

import java.util.*;

/**
 * 轮流分配的均衡器
 *
 * @author Payne 646742615@qq.com
 * 2018/12/24 21:40
 */
public class TurningBalancer implements Balancer {

    @Override
    public String getAlgorithm() {
        return "turning";
    }

    @Override
    public Map<String, List<JobSharding>> balance(Map<String, Executor.Information> executors, Task task) throws BalancingException {
        // 总分片数量
        int total = task.getShardingTotal();
        if (total <= 0) {
            throw new BalancingException("task sharding total must bigger than zero");
        }

        // 执行器数量
        int count = executors != null ? executors.size() : 0;
        if (count <= 0) {
            throw new BalancingException("no available job executors");
        }

        // 分片对照表
        @SuppressWarnings("UnstableApiUsage")
        Map<String, String> table = StrKit.isBlank(task.getShardingTable())
                ? Collections.<String, String>emptyMap()
                : Splitter.on(',')
                .trimResults()
                .withKeyValueSeparator('=')
                .split(task.getShardingTable());

        // 轮流分配
        Map<String, List<JobSharding>> map = new LinkedHashMap<>();
        String[] array = executors.keySet().toArray(new String[0]);
        for (int index = 0; index < total; index++) {
            String executor = array[index % count];
            String value = table.get(String.valueOf(index));
            JobSharding jobSharding = new JobSharding(task.getJobName(), task.getJobGroup(), index, task.getShardingTotal(), value);
            jobSharding.setJobName(task.getJobName());
            List<JobSharding> shardings = map.get(executor);
            if (shardings == null) {
                map.put(executor, shardings = new ArrayList<>());
            }
            shardings.add(jobSharding);
        }

        return map;
    }
}
